/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package studio.raptor.ddal.dashboard.rule;


import com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.ddal.common.algorithm.AbstractHashRangeShardAlgorithm;
import studio.raptor.ddal.common.algorithm.ShardValue;
import studio.raptor.ddal.common.algorithm.SingleKeyShardAlgorithm;
import studio.raptor.ddal.common.exception.GenericException;
import studio.raptor.ddal.common.exception.code.RouteErrCodes;

import java.io.IOException;
import java.util.*;

/**
 * 分片算法：分片字段为数字型，键值对方式
 * @author Sam
 * @since 3.0.0
 */
public class TestKeyValueSingleKeyShardAlgorithm implements SingleKeyShardAlgorithm<Comparable<?>> {

    private static Logger logger = LoggerFactory.getLogger(TestKeyValueSingleKeyShardAlgorithm.class);

    private static final String BASIC_DIR = "ddal/";
    private final String fileName;
    protected List<KeyValue> keyValueList = new ArrayList<>();
    protected Map<String, String> shardTarget = new HashMap<>();
    protected String keyvalueText;

    public TestKeyValueSingleKeyShardAlgorithm(String fileName) {
        this.fileName = fileName;
        readFile();
        parseRange();
        loadTargetMap();
    }

    @Override
    public String doEqual(Collection<String> shards, ShardValue<Comparable<?>> shardValue) {
        String value = String.valueOf(shardValue.getValue());
        String suffix = SEPARATOR + shardTarget.get("" + rangeShardValue(value));
        for (String shard : shards) {
            if (shard.endsWith(suffix)) {
                return shard;
            }
        }
        throw new UnsupportedOperationException();
    }

    @Override
    public Collection<String> doIn(Collection<String> shardings, ShardValue<Comparable<?>> shardValue) {
        Collection<String> r = new HashSet<>();
        for (Object value : shardValue.getValues()) {
            String suffix = SEPARATOR + shardTarget.get( "" + rangeShardValue(String.valueOf(value)));
            for (String shard : shardings) {
                if (shard.endsWith(suffix)) {
                    r.add(shard);
                }
            }
        }
        return r;
    }

    @Override
    public Collection<String> doBetween(Collection<String> shardings, ShardValue<Comparable<?>> shardValue) {
        throw new UnsupportedOperationException();
    }

    public int rangeShardValue(String shardValue) {
        for (KeyValue keyValue : keyValueList) {
            if (keyValue.isKey(Integer.parseInt(shardValue))) {
                return keyValue.getKey();
            }
        }
        throw new RuntimeException("No range found for " + shardValue);
    }

    /**
     * 读取文件中Range范围文本
     */
    private void readFile() {
        try {
            Properties hashRangeProperties = new Properties();
            hashRangeProperties.load( AbstractHashRangeShardAlgorithm.class.getClassLoader()
                    .getResourceAsStream( BASIC_DIR + fileName ) );

            String rangeText = hashRangeProperties.getProperty( "keyvalue" );
            if (null != rangeText && rangeText.length() > 0 && rangeText.indexOf( '-' ) > 0) {
                this.keyvalueText = rangeText;
            } else {
                throw new GenericException( RouteErrCodes.ROUTE_429, new Object[]{""} );
            }
        } catch (IOException ioe) {
            logger.error( "Read keyvalue prop failed.", ioe );
            throw new GenericException( RouteErrCodes.ROUTE_428 );
        }
    }

    /**
     * 解析Range范围
     *
     * 单字段分片时Range为一个List
     * 多字段分片时Range为多个List
     */
    private void parseRange() {
        keyValueList = createRanges(keyvalueText);
    }

    protected List<KeyValue> createRanges(String rangeContext){
        List<KeyValue> ranges = new ArrayList<>();
        Iterable<String> rangeMetaArray = Splitter.on(",").trimResults().split(rangeContext);
        for (String rangeMeta : rangeMetaArray) {
            List<String> rangeItem = Splitter.on("-").trimResults().splitToList(rangeMeta);
            if (rangeItem.size() == 2) {
                ranges.add(new KeyValue(Integer.parseInt(rangeItem.get(0)), Integer.parseInt(rangeItem.get(1))));
            } else {
                logger.error("Invalid range configuration.", keyvalueText);
                throw new GenericException( RouteErrCodes.ROUTE_429, new Object[]{rangeMeta});
            }
        }
        return ranges;
    }

    protected void loadTargetMap() {
        int targetIndex = 0;
        KeyValue keyValue=null;
        for(int j=0; j<keyValueList.size(); j++){
            keyValue=keyValueList.get( j );
            shardTarget.put(""+keyValue.getKey(), keyValue.getValue() +"");
        }
    }


    /**
     * 范围
     */
    public static class KeyValue {
        private int key;
        private int value;

        public int getKey() {
            return key;
        }

        public int getValue() {
            return value;
        }

        KeyValue(int key, int value) {
            this.key = key;
            this.value = value;
        }

        public boolean isKey(int key) {
            return this.key==key;
        }
    }
}
